package com.bjzspace.lwq.kafka.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class CustomProducerTransaction {

    public static void main(String[] args) {

        // 0.配置
        Properties properties = new Properties();
        // 链接集群
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "hadoop101:9092,hadoop102:9092,hadoop103:9092");
        // 指定key 和 value得序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        /**
         * 设置全局事务id, 全局唯一
         */
        properties.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "transaction_id_01");

        // 1.创建kafka生产者对象
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        // 初始化事务
        producer.initTransactions();
        // 开启事务
        producer.beginTransaction();
        try {
            // 2.发送消息
            for (int i = 0; i < 5; i++) {
                producer.send(new ProducerRecord<>("first", "liuwenqing" + i));
            }
            //提交事务
            producer.commitTransaction();
        }catch (Exception e) {
            // 释放事务
            producer.abortTransaction();
        }


        // 3.关闭资源
        producer.close();
    }

}
